package com.jianying.day07;

import com.jianying.beans.WaterSensor;
import com.jianying.func.WaterSensorMapFunction;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * ClassName: Flink05_TopN_1
 * Package: com.atguigu.day07
 * Description:
 *
 * @Author: tubage
 * @Create: 2024/4/8 14:13
 * @Version: 1.0
 */
public class Flink05_TopN_2 {
    public static void main(String[] args) throws Exception {
        // TODO top N 案例
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> wsDS = env.socketTextStream("localhost", 8888)
                .map(new WaterSensorMapFunction());

        // TODO 指定watermark
        SingleOutputStreamOperator<WaterSensor> withWaterMarkDS = wsDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<WaterSensor>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<WaterSensor>() {
                                    @Override
                                    public long extractTimestamp(WaterSensor waterSensor, long l) {
                                        return waterSensor.getTs() * 1000; // 事件时间字段 毫秒
                                    }
                                }
                        )
        );

        // TODO 按照水位值 分组
        KeyedStream<WaterSensor, Integer> vcKeyDS = withWaterMarkDS.keyBy(WaterSensor::getVc);
        // TODO 对分组中的数据开窗 每个水位值一个窗口
        WindowedStream<WaterSensor, Integer, TimeWindow> windowDS = vcKeyDS.window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)));
        // TODO 对窗口数据处理，增量+全量 计算当前窗口 水位值 出现次数
        SingleOutputStreamOperator<Tuple3<Integer, Integer, Long>> aggregateDS = windowDS.aggregate(
                new AggregateFunction<WaterSensor, Integer, Integer>() {
                    @Override
                    public Integer createAccumulator() {
                        return 0;
                    }

                    @Override
                    public Integer add(WaterSensor waterSensor, Integer accu) {
                        return accu + 1;
                    }

                    @Override
                    public Integer getResult(Integer accu) {
                        return accu;
                    }

                    @Override
                    public Integer merge(Integer integer, Integer acc1) {
                        return null;
                    }
                },

                new ProcessWindowFunction<Integer, Tuple3<Integer, Integer, Long>, Integer, TimeWindow>() {
                    @Override
                    public void process(Integer vc, Context context, Iterable<Integer> elements, Collector<Tuple3<Integer, Integer, Long>> out) throws Exception {
                        int count = elements.iterator().next();
                        long end = context.window().getEnd();
                        out.collect(Tuple3.of(vc, count, end));
                    }
                }
        );

        // TODO 按照窗口结束时间进行分组
        KeyedStream<Tuple3<Integer, Integer, Long>, Long> windowEndKeyedStream = aggregateDS.keyBy(t -> t.f2);
        
        // TODO 对分组后的数据进行处理
        SingleOutputStreamOperator<String> processDS = windowEndKeyedStream.process(
                new KeyedProcessFunction<Long, Tuple3<Integer, Integer, Long>, String>() {
                    Map<Long, List<Tuple3<Integer, Integer, Long>>> vcCountMap = new HashMap<>();

                    @Override
                    public void processElement(Tuple3<Integer, Integer, Long> value, Context ctx, Collector<String> out) throws Exception {
                        // TODO 处理 缓存到map里 （end窗口，List<Tuple3<水位值，次数，所属窗口>>）
                        Long windowEnd = value.f2;
                        if(vcCountMap.containsKey(windowEnd)) {
                           vcCountMap.get(windowEnd).add(value);
                        }else {
                            ArrayList<Tuple3<Integer, Integer, Long>> tuple3List = new ArrayList<>();
                            tuple3List.add(value);
                            vcCountMap.put(windowEnd,tuple3List);
                        }
                        // TODO 注册定时器（多个并行度下 等所有数据全了之后排序）
                        ctx.timerService().registerEventTimeTimer(windowEnd + 1);
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                        // TODO 排序之后逻辑在这执行
                        Long windowEnd = ctx.getCurrentKey();
                        List<Tuple3<Integer, Integer, Long>> tuple3List = vcCountMap.get(windowEnd);

                        tuple3List.sort((t1,t2) -> t2.f1 - t1.f1);

                        int topN = Math.min(2, tuple3List.size());
                        StringBuilder outStr = new StringBuilder();
                        outStr.append("================================\n");

                        for (int i = 0; i < topN; i++) {
                            Tuple3<Integer, Integer, Long> vcCountTuple3 = tuple3List.get(i);
                            outStr.append("Top" + (i + 1) + "\n");
                            outStr.append("vc=" + vcCountTuple3.f0 + "\n");
                            outStr.append("count=" + vcCountTuple3.f1 + "\n");
                            outStr.append("窗口结束时间=" + vcCountTuple3.f2 + "\n");
                            outStr.append("================================\n");
                        }

                        tuple3List.clear();
                        out.collect(outStr.toString());



                    }
                }
        );

        processDS.print();

        env.execute();
    }
}
